package PassengerVehicle

import java.util.Properties

import PassengerVehicle.ylpzs.readFile
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}

object CalFactors {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.WARN)
    val cdict = readFile("D:/projects/spark/src/main/resource/incity.txt")
    val pdict = readFile("D:/projects/spark/src/main/resource/inprovince.txt")
    //创建sparkSession
    val spark: SparkSession = SparkSession.builder().appName("FactorsInOut").master("local[4]").config("spark.debug.maxToStringFields", "1000").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    //创建properties对象 设置连接mysql的信息
    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")
    //读取mysql数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql?useUnicode=true&characterEncoding=UTF-8", "zhjg_jsy_ticket_smz_curmonth", prop)
    val vehicleDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/localsql?useUnicode=true&characterEncoding=UTF-8", "zhjg_zy_vehicle", prop)
    import spark.sql
    //过滤订单表，查询车牌号空和鲁F、鲁Y的车辆
    mysqlDF.filter("SCKCCPHM ='' or SCKCCPHM like '鲁F%' or SCKCCPHM like '鲁Y%'").createOrReplaceTempView("passenger")
    //查询车辆状态为营运的
    vehicleDF.filter("VEH_STATUS = 10").createOrReplaceTempView("vehicle")
    val mysqlDataFrame = sql("select XL,FCRQSJ,SCKCCPHM,DDZ from passenger ").filter("FCRQSJ like '2019-04%'").createOrReplaceTempView("data")
    /*    //获取4月份的订单数据:查询实际载客量，并分4个维度的
        val testMonthData = sql("select XL,FCRQSJ,SCKCCPHM,DDZ from passenger where FCRQSJ like '2019-04%'")*/
    //查询核定载客量：区分4个维度
    sql("select SCKCCPHM,XL,FCRQSJ,first(DDZ) as DDZ from data GROUP BY XL,SCKCCPHM,FCRQSJ").filter("FCRQSJ like '2019-04%'").createOrReplaceTempView("a")
    sql("select a.XL, a.SCKCCPHM,first(a.DDZ) as DDZ,count(1) as checi,if(a.SCKCCPHM = '',50,a.SCKCCPHM) as seat from a group by XL,SCKCCPHM").createOrReplaceTempView("ttt")
    sql("select a.XL, a.SCKCCPHM,first(a.DDZ) as DDZ,first(a.checi) as checi,if(a.SCKCCPHM='' or a.SCKCCPHM is null,50,first(b.SEAT_COUNT)) as seat from ttt a left join vehicle b on a.SCKCCPHM = b.VEH_NO group by XL,SCKCCPHM").createOrReplaceTempView("sss")
    //查询线路、车牌号、到达站、车次、座位数、运力（车次*座位数）
    val ylcs = sql("select XL,SCKCCPHM,DDZ,checi,seat,if(seat is null,0,seat*checi) as yl from sss").createOrReplaceTempView("t1")
    import spark.implicits._
    val sjzkl = sql("select XL from data").groupBy("XL").count().withColumn("sjzkl", $"count").createOrReplaceTempView("t2")
    val fData = sql("select t1.XL,t1.DDZ,t1.yl,t2.sjzkl from t1,t2 where t1.XL = t2.XL").show(500)
    //YlInCity(ylcs)

    /*    //实际载客量
        InCity(testMonthData)
        OutCity(testMonthData)
        OutProvince(testMonthData)*/
    spark.stop()

    //运力：市内
    def YlInCity(s: DataFrame): Unit = {
      // InCity市内，OutCity，
      val result = s.rdd.map(item => {
        val xl = item.getString(0)
        val ddz = item.getString(2)
        val yl = item.get(4)
        if (matchDDZ(xl, ddz, cdict)) {
          ("InCity:", yl.toString.toInt)
        } else {
          ("OutCity:", 0)
        }
      }).reduceByKey((x, y) => x + y)

      result.foreach(x => {
        println(x._1 + ":" + x._2)
      })
    }

    //运力：跨市
    def YlOutCity(d: DataFrame): Unit = {
      val outCityValue = d.rdd.filter(row => {
        val xl = row.getString(0)
        val ddz = row.getString(2)
        val yl = row.get(4)
        !matchDDZ(xl, ddz, cdict) && matchDDZ(xl, ddz, pdict)
      })
      var sumYl = 0
      outCityValue.collect().foreach(r => {
        sumYl = sumYl + r.get(4).toString.toInt
      })
      println("跨市运力:" + sumYl)
    }

    //运力：跨省
    def YlOutProvince(d: DataFrame): Unit = {

    }


    //实际载客量:市内
    def InCity(s: DataFrame): Unit = {
      // InCity市内，OutCity，
      val result = s.rdd.map(item => {
        val xl = item.getString(0)
        val ddz = item.getString(2)
        if (matchDDZ(xl, ddz, cdict)) {
          ("InCity", 1)
        } else {
          ("OutCity", 1)
        }
      }).reduceByKey((x, y) => x + y)

      result.foreach(x => {
        println(x._1 + ":" + x._2)
      })
    }

    //实际载客量:跨市
    def OutCity(s: DataFrame): Unit = {
      val outCityValue = s.rdd.filter(row => {
        val xl = row.getString(0)
        val ddz = row.getString(2)
        !matchDDZ(xl, ddz, cdict) && matchDDZ(xl, ddz, pdict)
      })
      println("跨省:" + outCityValue.count())
    }

    //实际载客量:跨省
    def OutProvince(s: DataFrame): Unit = {
      val outProvinceValue = s.rdd.filter(row => {
        val xl = row.getString(0)
        val ddz = row.getString(2)
        !matchDDZ(xl, ddz, pdict)
      }).count()
      println("跨省:" + outProvinceValue)
    }

    def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
      val reg1 = "[0-9]"
      val reg2 = "（[\\u4e00-\\u9fa5]*）"
      val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
      val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口|总站|南站|北站|东站|西站|汽车"

      val xllen = xl.split("-").length
      if (dict.indexOf(ddz.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else if (xllen == 3 && dict.indexOf(xl.split("-")(2).replaceAll(reg1, "").replaceAll(reg2, "").replaceAll(reg3, "").replaceAll(reg4, "").trim) > -1) {
        true
      } else {
        false
      }
    }
  }
}
